package com.csw.spark

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

//3、统计每科都及格的学生 [学号，姓名，班级，科目，分数]
object Text5 {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf()
      .setMaster("local")
      .setAppName("text1")

    val sc: SparkContext = new SparkContext(conf)

    //读取数据
    val scoresRDD: RDD[String] = sc.textFile("spark/data/score.txt")

    val studentsRDD: RDD[String] = sc.textFile("spark/data/students.txt")

    //获取id和score的kv集合
    val scoresIdScoreRDD: RDD[(String, Int)] = scoresRDD.map(i => (i.split(",")(0), i.split(",")(2).toInt))

    //对scoresIdScoreRDD进行分组
    val scoresGroupRDD: RDD[(String, Iterable[Int])] = scoresIdScoreRDD.groupByKey()

    //对分组后的rdd进行map得到id和6门学科成绩
    val scoresGroupMapRDD: RDD[(String, Int, Int, Int, Int, Int, Int)] = scoresGroupRDD.map(kv => {
      val id: String = kv._1
      val list: List[Int] = kv._2.toList
      val score1: Int = list(0)
      val score2: Int = list(1)
      val score3: Int = list(2)
      val score4: Int = list(3)
      val score5: Int = list(4)
      val score6: Int = list(5)
      (id, score1, score2, score3, score4, score5, score6)
    })

    //过滤得到6门科目都及格的学生
    val scoresGroupMapFilterRDD: RDD[(String, Int, Int, Int, Int, Int, Int)] = scoresGroupMapRDD.filter(i => {
      i._2 > 60
      i._3 > 60
      i._4 > 60
      i._5 > 60
      i._6 > 60
      i._7 > 60
    })

    //获取6门科目都及格的学生id
    val getId: RDD[String] = scoresGroupMapFilterRDD.map(i => {
      val id: String = i._1
      id
    })
    val getid: Array[String] = getId.collect()

    //对成绩表进行转换
    val scoresMapRDD: RDD[(String, String)] = scoresRDD.map(i => {
      val id: String = i.split(",")(0)
      val subjectId: String = i.split(",")(1)
      val score: String = i.split(",")(2)

      (id, subjectId + "," + score)
    })
    //过滤得到6门科目全部及格的id,科目，成绩
    val scoresMapFilterRDD: RDD[(String, String)] = scoresMapRDD.filter(kv => {
      val id: String = kv._1
      getid.contains(id)
    })

    //先对student表做处理，取出要显示的数据和学生表关联总分大于年级平均分的学生
    val studentsMapRDD: RDD[(String, String)] = studentsRDD.map(i => {
      val id: String = i.split(",")(0)
      val name: String = i.split(",")(1)
      val clazz: String = i.split(",")(4)
      (id, name + "," + clazz)
    })
    val result: RDD[(String, (String, Option[String]))] = scoresMapFilterRDD.leftOuterJoin(studentsMapRDD)

    result.foreach(println)
    println(result.count())
  }
}
